package skewreduce.framework.physical;

import java.io.IOException;
import java.util.Arrays;
import java.util.Comparator;
import java.util.List;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskInputOutputContext;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.util.StringUtils;

import skewreduce.framework.logical.LogicalOp;


public abstract class PMergeOp extends MonitoredPhysicalOp {
    private final static Log LOG = LogFactory.getLog("PLAN");

    private boolean mapOnly;

    public static class MergeInputFormat<K,V> extends SequenceFileInputFormat<K,V> {
        @Override
        protected final boolean isSplitable(JobContext context,Path filename) {
            return false;
        }
    }

    @Override
    protected void setupInput(Job job,List<Path> input) throws IOException {
        mapOnly = job.getConfiguration().getBoolean("skewreduce.merge.mapOnly",false);

        if ( LOG.isDebugEnabled() ) {
            LOG.debug(String.format("Map only = %b,input size = %d",mapOnly,input.size()));
        }

        if ( mapOnly && input.size() > 1 ) {
            // sort input in order of size
            FileSystem fs = plan.getDriver().getFileSystem();
            FileStatus[] stats = new FileStatus[input.size()];
            int i = 0;
            for ( Path p : input ) {
                stats[i++] = fs.getFileStatus(p);
            }
            Arrays.sort(stats,new Comparator<FileStatus>() {
                @Override
                public int compare(FileStatus a,FileStatus b) {
                    return Long.signum(b.getLen() - a.getLen());
                }
            });

            StringBuilder dirStr = new StringBuilder();
            for ( i = 0; i < stats.length-1; ++i ) {
                Path p = stats[i].getPath().makeQualified(fs);
                if ( dirStr.length() > 0 )
                    dirStr.append(',');
                dirStr.append(StringUtils.escapeString(p.toString()));
            }
            FileInputFormat.addInputPath(job,stats[i].getPath());
            job.getConfiguration().set("skewreduce.merge.extraInput",dirStr.toString());

            if ( LOG.isDebugEnabled() ) {
                LOG.debug("Main input to merge = "+stats[i].getPath());
                LOG.debug("Extra input to merge = "+dirStr.toString());
            }
        } else {
            super.setupInput(job,input);
        }
    }

    protected Path getOutputByContext(PhysicalOp op) {
        LogicalOp.Type optype = op.getType();
        String fn = ( mapOnly ) ? "part-m-00000" : "part-r-00000";
        if ( optype != LogicalOp.Type.MERGE 
                && optype != LogicalOp.Type.FINALIZE ) {
            throw new IllegalArgumentException();
        }

        Path p = new Path(getOutputPath(),fn);
        return p;
    }

    public static interface MergeDataIterator<KEY,VALUE> {
        public KEY getCurrentKey() throws IOException, InterruptedException;
        public VALUE getCurrentValue() throws IOException, InterruptedException;
        public boolean nextKeyValue() throws IOException, InterruptedException;
        public void close() throws IOException, InterruptedException;
    }

    public static class TaskContextDataIterator<K,V> implements MergeDataIterator<K,V> {
        private final TaskInputOutputContext<K,V,?,?> context;

        public TaskContextDataIterator(TaskInputOutputContext<K,V,?,?> c) {
            this.context = c;
        }

        public K getCurrentKey() throws IOException, InterruptedException {
            return context.getCurrentKey();
        }
        public V getCurrentValue() throws IOException, InterruptedException {
            return context.getCurrentValue();
        }
        public boolean nextKeyValue() throws IOException, InterruptedException {
            return context.nextKeyValue();
        }
        public void close() throws IOException, InterruptedException {}
    }

    public static class RecordReaderDataIterator<K,V> implements MergeDataIterator<K,V> {
        private final RecordReader<K,V> reader;

        public RecordReaderDataIterator(RecordReader<K,V> reader) {
            this.reader = reader;
        }

        public K getCurrentKey() throws IOException, InterruptedException {
            return reader.getCurrentKey();
        }
        public V getCurrentValue() throws IOException, InterruptedException {
            return reader.getCurrentValue();
        }
        public boolean nextKeyValue() throws IOException, InterruptedException {
            return reader.nextKeyValue();
        }
        public void close() throws IOException, InterruptedException {
            reader.close();
        }
    }
}
